package com.runssnail.demo.kafka.jms.spring;

import com.runssnail.demo.kafka.jms.KafkaTopicPublisher;
import org.apache.commons.lang3.Validate;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;

import javax.jms.TopicPublisher;
import java.util.Properties;

/**
 * TopicPublisher FactoryBean
 *
 * <p/>
 * Created by zhengwei on 2015/8/28.
 */
public class TopicPublisherFactoryBean implements FactoryBean<TopicPublisher>, InitializingBean, DisposableBean {

    private TopicPublisher topicPublisher;

    /**
     * kafka配置
     */
    private Properties configs;

    @Override
    public TopicPublisher getObject() throws Exception {
        return this.topicPublisher;
    }

    @Override
    public Class<?> getObjectType() {
        return TopicPublisher.class;
    }

    @Override
    public boolean isSingleton() {
        return true;
    }

    @Override
    public void afterPropertiesSet() throws Exception {

        Validate.notEmpty(this.configs, "configs is required");

        KafkaProducer kafkaProducer = new KafkaProducer<String, Object>(configs);

        KafkaTopicPublisher publisher = new KafkaTopicPublisher(kafkaProducer);
        this.topicPublisher = publisher;
    }

    @Override
    public void destroy() throws Exception {
        this.topicPublisher.close();
    }

    /**
     * kafka配置
     *
     * @param configs kafka配置
     */
    public void setConfigs(Properties configs) {
        this.configs = configs;
    }
}
